package rpc

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"net"
	"platon-tools/go-service/event"
	go_logger "platon-tools/go-service/go-logger"
	"platon-tools/go-service/protocol/rpc_pb"
	"platon-tools/go-service/service"
	"sync"
	"time"
)

const (
	eventChSize    = 100
	version        = 101
	connectTimeout = 30 * time.Second
)

type RegisterMonitor struct {
	url   string
	errCh chan error
}

type CenterRcp struct {
	monitors struct {
		m map[string]rpc_pb.MonitorServiceClient
		sync.RWMutex
	}

	service *service.Service

	ctx    context.Context
	cancel context.CancelFunc

	errCh   chan error
	regisCh chan RegisterMonitor
	eventCh chan interface{}

	url string

	lister net.Listener
}

func (cr *CenterRcp) Ping(ctx context.Context, request *rpc_pb.RpcRequest) (*rpc_pb.ResultRequest, error) {
	return &rpc_pb.ResultRequest{
		Code:    0,
		Err:     nil,
		Version: version,
	}, nil
}

func (cr *CenterRcp) RegisterMonitor(ctx context.Context, request *rpc_pb.RegisterRequest) (*rpc_pb.ResultRequest, error) {
	if request.Version > version {
		return nil, errors.New(fmt.Sprintf("Version is too high current => [%d], got => [%d]", version, request.Version))
	}

	go_logger.Logger.Sugar().Debug("monitor register quest", request.Url)

	errCh := make(chan error)
	cr.regisCh <- RegisterMonitor{
		url:   request.Url,
		errCh: errCh,
	}
	defer close(errCh)

	select {
	case err, ok := <-errCh:
		if ok && err != nil {
			return nil, err
		}
	case <-time.After(connectTimeout):
		return nil, errors.New("connecting time out")
	}

	go_logger.Logger.Sugar().Debug("monitor register ok", request.Url)

	return &rpc_pb.ResultRequest{
		Code:    0,
		Err:     nil,
		Version: version,
	}, nil
}

func (cr *CenterRcp) Push(ctx context.Context, request *rpc_pb.PushRequest) (*rpc_pb.ResultRequest, error) {
	if request.Version > version {
		return nil, errors.New(fmt.Sprintf("Version is too high current => [%d], got => [%d]", version, request.Version))
	}

	rr := rpc_pb.ResultRequest{
		Code:    0,
		Err:     nil,
		Version: version,
	}

	switch request.Type {
	case rpc_pb.MsgType_PodsList:
		pods := event.PodsList{}
		if err := json.Unmarshal(request.Data, &pods); err != nil {
			return nil, err
		}
		cr.service.GetEvent().Send(event.Msg{
			Type:    rpc_pb.MsgType_PodsList,
			Payload: pods,
		})
	case rpc_pb.MsgType_Validator:
		validator := event.ValidatorList{}
		if err := json.Unmarshal(request.Data, &validator); err != nil {
			return nil, err
		}
		cr.service.GetEvent().Send(event.Msg{
			Type:    rpc_pb.MsgType_Validator,
			Payload: validator,
		})
	case rpc_pb.MsgType_Verifier:
		validator := event.VerifierList{}
		if err := json.Unmarshal(request.Data, &validator); err != nil {
			return nil, err
		}
		cr.service.GetEvent().Send(event.Msg{
			Type:    rpc_pb.MsgType_Verifier,
			Payload: validator,
		})
	case rpc_pb.MsgType_NodeMonitorItem:
		nodeMonitor := event.NodeMonitorItem{}
		if err := json.Unmarshal(request.Data, &nodeMonitor); err != nil {
			return nil, err
		}
		go_logger.Logger.Sugar().Debugf("MsgType_NodeMonitorItem %v", nodeMonitor)
		cr.service.GetEvent().Send(event.Msg{
			Type:    rpc_pb.MsgType_NodeMonitorItem,
			Payload: nodeMonitor,
		})
	case rpc_pb.MsgType_NewPod:
		npod := event.NewPod{}
		if err := json.Unmarshal(request.Data, &npod); err != nil {
			return nil, err
		}
		go_logger.Logger.Sugar().Debugf("MsgType_NewPod %v", npod)
		cr.service.GetEvent().Send(event.Msg{
			Type:    rpc_pb.MsgType_NodeMonitorItem,
			Payload: npod,
		})

	}

	return &rr, nil
}

func (cr *CenterRcp) Start(ctx context.Context, config *service.Config, s *service.Service) error {
	self, cancel := context.WithCancel(ctx)
	cr.url = config.RpcURL
	cr.service = s
	cr.ctx = self
	cr.cancel = cancel
	cr.eventCh = make(chan interface{}, eventChSize)
	cr.regisCh = make(chan RegisterMonitor)
	cr.monitors.m = map[string]rpc_pb.MonitorServiceClient{}

	go cr.runLoop()
	go cr.rpcLoop()

	return nil
}

func (cr *CenterRcp) runLoop() {
	defer cr.cancel()
	go_logger.Logger.Debug("center rcp run loop")

	// 防止发送event时没有订阅者，发生异常。
	go func() {
		ch := make(chan event.Msg, 5)
		defer close(ch)

		sub := cr.service.GetEvent().Subscribe(ch)

		defer sub.Unsubscribe()
		//defer sub2.Unsubscribe()
		for {
			select {
			case <-cr.ctx.Done():
				return
			case <-sub.Err():
				return
			case msg, ok := <-ch:
				if ok {
					switch msg.Type {
					case rpc_pb.MsgType_PodsList:
						if pods, ok := msg.Payload.(event.PodsList); ok {
							go_logger.Logger.Sugar().Debugf("订阅消息 pods list【%+v】", pods)
						}
					case rpc_pb.MsgType_Validator:
						if vlist, ok := msg.Payload.(event.ValidatorList); ok {
							go_logger.Logger.Sugar().Debugf("订阅消息 区块奖励【%f】", vlist.MiningReward)
						}
					case rpc_pb.MsgType_Verifier:
						if vlist, ok := msg.Payload.(event.VerifierList); ok {
							go_logger.Logger.Sugar().Debugf("订阅消息 质押奖励【%f】", vlist.StakingReward)
						}
					case rpc_pb.MsgType_NodeMonitorItem:
						if item, ok := msg.Payload.(event.NodeMonitorItem); ok {
							go_logger.Logger.Sugar().Debugf("订阅消息 节点监控 数量【%s】", item.NodeName)
						}

					}

				}

			}
		}
	}()
	for {
		select {
		case <-cr.ctx.Done():
			return
		case err, ok := <-cr.errCh:
			if ok {
				go_logger.Logger.Error("runLoop exit", zap.Error(err))
				return
			}
		case r, ok := <-cr.regisCh:
			if ok {
				go cr.generateMonitorClient(r.url, r.errCh)
			}
		}
	}
}

func (cr *CenterRcp) rpcLoop() {
	defer cr.cancel()
	go_logger.Logger.Debug("rpcLoop rcp run loop", zap.String("url", cr.url))
	lis, err := net.Listen("tcp", cr.url)
	if err != nil {
		cr.errCh <- err
		return
	}

	cr.lister = lis

	grpcServer := grpc.NewServer()
	rpc_pb.RegisterCentreServiceServer(grpcServer, cr)
	err = grpcServer.Serve(lis)
	if err != nil {
		go_logger.Logger.Error("center rpc error", zap.Error(err))
	}
}

func (cr *CenterRcp) Close() error {
	if cr.IsRunning() {
		cr.cancel()
		cr.lister.Close()
		close(cr.errCh)
		close(cr.eventCh)
		close(cr.regisCh)
	}

	return nil
}
func (cr *CenterRcp) IsRunning() bool {
	select {
	case <-cr.ctx.Done():
		return false
	default:
		return true
	}
}
func (cr *CenterRcp) Name() string {
	return "CENTERRPC"
}

func (cr *CenterRcp) generateMonitorClient(url string, errCh chan error) {

	writeCh := func(err error) {
		select {
		case <-errCh:
		default:
			errCh <- err
		}
	}

	conn, err := grpc.DialContext(cr.ctx, url, grpc.WithInsecure())
	if err != nil {
		writeCh(err)
		go_logger.Logger.Error("Dial monitor failed", zap.Error(err), zap.String("monitor url", url))
		return
	}

	client := rpc_pb.NewMonitorServiceClient(conn)

	ctx, _ := context.WithTimeout(cr.ctx, 10*time.Second)
	if list, err := client.GetPodsList(ctx, &rpc_pb.RpcRequest{Version: version}); err == nil {
		go_logger.Logger.Sugar().Debugf("pod list [%+v]", string(list.Pods))
	}

	cr.monitors.Lock()
	defer cr.monitors.Unlock()
	cr.monitors.m[url] = client
	writeCh(nil)
}
